package com.you.raft.core.impl;

import com.you.raft.core.*;
import com.you.raft.core.task.ElectionTask;
import com.you.raft.core.task.HeartBeatTask;
import com.you.raft.rpc.RpcRaftClient;
import com.you.raft.common.entity.Request;
import com.you.raft.common.entity.*;
import com.you.raft.rpc.RpcRaftServer;
import com.you.raft.rpc.impl.DefaultRpcClient;
import com.you.raft.rpc.impl.DefaultRpcServer;
import com.you.raft.thread.RaftThreadPool;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.concurrent.*;

@Getter
@Setter
@Slf4j
public class DefaultRaftNode implements RaftNode {

    /* ============ 定时任务线程 ============ */
    private ElectionTask electionTask;

    private HeartBeatTask heartBeatTask;

    /* ============ RPC框架 ============== */
    private RpcRaftServer rpcServer;

    private RpcRaftClient rpcClient;

    /* ============= 基本模块 ============== */
    /**
     * 一致性模块实现
     */
    private Consensus consensus;
    /**
     * 日志条目集；每一个条目包含一个用户状态机执行的指令，和收到时的任期号
     */
    private LogModule logModule;
    /**
     * 状态机
     */
    private StateMachine stateMachine;
    /**
     * 客户端服务
     */
    private ClientService clientService;

    /* ======================================== */
    /* ============ 节点基本状态 ============ */
    private volatile int status = NodeStatus.FOLLOWER;

    private volatile boolean running = false;

    private Cluster cluster;

    /* ============ 基本参数 ============ */
    /**
     * 上一次选举时间
     */
    private volatile long preElectionTime = 0;
    /**
     * 上次一心跳时间戳
     */
    private volatile long preHeartBeatTime = 0;

    /**
     * 当前任期号
     */
    private volatile long currentTerm = 1;
    /**
     * 当前节点投票给
     */
    private volatile String votedFor;

    /**
     * 已经提交的日志最大索引值(日志复制到一半以上的节点)(领导者直接修改,追随者根据收到的领导者心跳修改)
     */
    private volatile long commitIndex = 0;

    /**
     * 被应用到状态机的日志最大索引值
     */
    private volatile long lastApplied = 0;

    /* ========== 领导人维护 ================== */

    /**
     * 需要发送给某个节点的下一个日志条目的索引值（对应value=已经发送的索引号+1）
     */
    private volatile Map<Node, Long> nextIndexs;

    /**
     * 对于每一个服务器，已经复制给他的日志的最高索引值
     */
    private volatile Map<Node, Long> matchIndexs;

    private static class DefaultRaftNodeLazyHolder {
        private static DefaultRaftNode INSTANCE = new DefaultRaftNode();
    }

    public static DefaultRaftNode getInstance() {
        return DefaultRaftNodeLazyHolder.INSTANCE;
    }


    @Override
    public void init(NodeConfig config) throws Exception {
        //初始化节点信息
        this.cluster = new Cluster();
        String selfAddress = "localhost:" + config.getSelfPort();
        cluster.setSelf(new Node(selfAddress));
        for (String address : config.getAllNodeAddresses()) {
            cluster.addNode(address);
        }
        //启动RPC服务器与客户端
        rpcServer = new DefaultRpcServer(config.getSelfPort(), this);
        rpcClient = new DefaultRpcClient();
        rpcServer.init();
        rpcClient.init();
        //加载模块
        logModule = DefaultLogModule.getInstance();
        stateMachine = DefaultStateMachine.getInstance();
        consensus = new DefaultConsensus(this);
        clientService = new DefaultClientService(this);
        //开启定时任务
        heartBeatTask = new HeartBeatTask(this);
        electionTask = new ElectionTask(this);
        RaftThreadPool.scheduleWithFixedDelay(heartBeatTask, 500);
        RaftThreadPool.scheduleAtFixedRate(electionTask, 5000, 500);
        //设置状态
        running = true;
        log.info("节点【{}】成功启动！", cluster.getSelf());
    }

    @Override
    public void destory() throws Exception {
        //销毁RPC服务器，客户端
        rpcClient.destory();
        rpcServer.destory();
        running = false;
        //销毁当前状态机
        log.info("节点销毁成功");
    }

    /**
     * 处理请求
     */
    @Override
    public VoteResponse handlerVoteRequest(VoteRequest request) {
        log.info("handler 投票请求:{}", request);
        return consensus.requestVote(request);
    }

    @Override
    public AppendEntryResponse handlerAppendEntry(AppendEntryRequest request) {
        log.info("handler 处理附加日志请求:{}", request);
        return consensus.appendEntry(request);
    }

    @Override
    public ClientResponse handlerClientRequest(ClientRequest request) {
        log.info("handler 处理客户端请求:{}", request);
        return clientService.clientRequest(request);
    }

    /**
     * 复制日志LogEntry到节点Node
     */
    public Future<Boolean> replication(Node cnode, LogEntry entry) {

        return RaftThreadPool.submit(() -> {
            long start = System.currentTimeMillis(), end = start;
            //20秒内可重试多次
            while (end - start < 20 * 1000L) {
                AppendEntryRequest entryRequest = new AppendEntryRequest();
                entryRequest.setTerm(currentTerm);
                entryRequest.setLeaderId(cluster.getSelf().getAddress());
                entryRequest.setLeaderCommit(commitIndex);
                long nextIndex = nextIndexs.get(cnode);
                LinkedList<LogEntry> logEntries = new LinkedList<>();
                if (entry.getIndex() >= nextIndex) {
                    // 复制从nextIndex到当前entry日志索引之间的多条日志
                    for (long i = nextIndex; i <= entry.getIndex(); i++) {
                        LogEntry l = logModule.get(i);
                        if (l != null) {
                            logEntries.add(l);
                        }
                    }
                } else {
                    //如果日志在nextIndex之前，仅复制entry
                    logEntries.add(entry);
                }
                //得到该日志的前一个日志-以设置附加日志请求的前一个信息信息
                LogEntry preLog = logModule.get(entry.getIndex() - 1);
                entryRequest.setPreLogTerm(preLog.getTerm());
                entryRequest.setPrevLogIndex(preLog.getIndex());
                entryRequest.setEntries(logEntries.toArray(new LogEntry[0]));


                Request request = Request.builder()
                        .cmd(Request.APPEND_ENTRIES)
                        .obj(entryRequest)
                        .addr(cnode.getAddress())
                        .build();
                try {
                    AppendEntryResponse result = rpcClient.send(request);
                    if (result == null) {
                        return false;
                    }
                    if (result.isSuccess()) {
                        log.info("追加日志请求成功, follower=[{}], entry=[{}]", cnode, entryRequest.getEntries());
                        // 更新日志两个追踪值
                        nextIndexs.put(cnode, entry.getIndex() + 1);
                        matchIndexs.put(cnode, entry.getIndex());
                        return true;
                    } else {
                        // 失败的原因一:对方任期更大
                        if (result.getTerm() > currentTerm) {
                            log.warn("对方{}任期{}比当前节点任期{}更大，当前节点【LEADER】->【FOLLOWER】",
                                    cnode, result.getTerm(), currentTerm);
                            currentTerm = result.getTerm();
                            status = NodeStatus.FOLLOWER;
                            return false;
                        } //失败的原因二: 对方前一个日志对不上导致拒绝此日志（往前找到第一个日志的前一个日志号和任期与对方最后一个日志号和任期相同）
                        else {
                            nextIndexs.put(cnode, nextIndex - 1 >= 1 ? nextIndex - 1 : 1);
                            log.warn("对方节点 {} 找不到前一个日志号与任期匹配的日志: [{}]", cnode.getAddress(),
                                    nextIndex);
                        }
                    }
                    end = System.currentTimeMillis();
                } catch (Exception e) {
                    log.warn(e.getMessage(), e);
                    return false;
                }
            }
            return false;
        });
    }


}
